package com.galeno.day10;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.io.File;
import java.io.FileOutputStream;
import java.io.FileWriter;
import java.io.Writer;
import java.util.concurrent.TimeUnit;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2719:18
 */
public class AsyncIODemo1 {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 12345);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

        DataStreamSource<String> lines = env.socketTextStream("192.168.77.3", 9999);
        //使用异步IO进行查询关联信息
        //unorderedWait发送请求返回的结果与发送请求的先后顺序可以一致
        //orderedWait发送请求返回的结果与发送请求的先后顺序一致
        SingleOutputStreamOperator<OrderBean> res = AsyncDataStream.unorderedWait(lines, new AsyncHttpQueryFunction(), 2000, TimeUnit.SECONDS, 10);
        System.out.println("-----------------------------------------------------------------------");
        // System.out.println(env.getExecutionPlan());

        // Writer fileWriter = new FileWriter(new File("D:/b.txt"),true);
        //   fileWriter.write(env.getExecutionPlan());
        // fileWriter.flush();


        res.print();

        env.execute();

    }
}
